package wsz.work.service.impl;

import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import wsz.work.service.PositionService;
import wsz.work.util.DBHelper;

import java.io.IOException;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;

/**
 * @author wsz
 * @date 2022/3/28 16:40
 **/
@Slf4j
@Service
public class PositionServiceImpl implements PositionService {

    @Autowired
    private RestHighLevelClient client;
    private static final String POSITION_INDEX = "position";

    @Override
    public List<Map<String, Object>> searchPos(String keyword, int pageNo, int pageSize) throws IOException {
        if (pageNo <= 1){
            pageNo = 1;
        }
        // 搜索
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        // 分页设置
        searchSourceBuilder.from((pageNo-1)*pageSize)
                .size(pageSize)
                .timeout(new TimeValue(60,TimeUnit.SECONDS));
        if (StringUtils.hasText(keyword)) {
            QueryBuilder builder = QueryBuilders.matchQuery("positionName",keyword);
            searchSourceBuilder.query(builder);
        }
        // 执行搜索
        SearchRequest searchRequest = new SearchRequest(POSITION_INDEX)
                .source(searchSourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest,RequestOptions.DEFAULT);
        SearchHit[]  hits = searchResponse.getHits().getHits();

        ArrayList<Map<String,Object>>  list = new ArrayList<>(hits.length);

        for (SearchHit hit:hits){
            list.add(hit.getSourceAsMap());
        }
        return list;
    }

    @Override
    public void importAll() {
        writeMySQLDataToES(POSITION_INDEX);
    }

    private String DEFAULT_STRING_1 = "美女多";
    private String DEFAULT_STRING_2 = "员工福利好";
    private String DEFAULT_STRING_3 = "美女多、员工福利好";

    /**
     * 额外补充的数据
     * @param size
     * @return
     * @throws IOException
     */
    @Override
    public List<Map<String, Object>> searchDefaultAdd(int size) throws IOException {
        if (size <= 1){
            return new ArrayList<>();
        }


        /**
         * 1.should [match,match],minimumShowldMatch
         */
//        QueryBuilder builder = QueryBuilders.boolQuery()
//                .should(QueryBuilders.matchQuery("positionAdvantage", DEFAULT_STRING_1))
//                .should(QueryBuilders.matchQuery("positionAdvantage", DEFAULT_STRING_2))
//                .should(QueryBuilders.matchQuery("positionAdvantage", DEFAULT_STRING_3))
//                .minimumShouldMatch(1);

        /**
         * matchQuery
         */
        QueryBuilder builder = QueryBuilders.matchQuery("positionAdvantage", DEFAULT_STRING_3);

        // 最前面size个数据
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder()
                .from(1)
                .size(size)
                .timeout(new TimeValue(60,TimeUnit.SECONDS))
                .query(builder);

        // 执行搜索
        SearchRequest searchRequest = new SearchRequest(POSITION_INDEX)
                .source(searchSourceBuilder);

        SearchResponse searchResponse = client.search(searchRequest,RequestOptions.DEFAULT);
        SearchHit[] hits = searchResponse.getHits().getHits();

        ArrayList<Map<String,Object>> list = new ArrayList<>(hits.length);

        for (SearchHit hit:hits){
            list.add(hit.getSourceAsMap());
        }
        return list;
    }

    /**
     * 数据导入
     * @param tableName
     */
    private void writeMySQLDataToES(String tableName) {
        BulkProcessor  bulkProcessor  = getBulkProcessor(client);

        Connection connection = null;
        PreparedStatement ps = null;
        ResultSet rs = null;
        try {
            connection = DBHelper.getConn();
            String  sql = "select * from " + tableName;
            ps = connection.prepareStatement(sql,ResultSet.TYPE_FORWARD_ONLY,ResultSet.CONCUR_READ_ONLY);

            // 每次20条数据
            ps.setFetchSize(20);
            rs = ps.executeQuery();
            ResultSetMetaData colData = rs.getMetaData();
            ArrayList<HashMap<String,String>> dataList = new ArrayList<>();
            HashMap<String,String> map;
            int  count = 0;
            // c 就是列的名字   v 就是列对应的值
            String  c = null;
            String  v = null;
            while (rs.next()) {
                count ++;
                map = new HashMap<>(64);
                for (int i=1; i< colData.getColumnCount(); i++){
                    c = colData.getColumnName(i);
                    v = rs.getString(c);
                    map.put(c,v);
                }
                dataList.add(map);
                // 每1万条 写一次   不足的批次的数据 最后一次提交处理
                if (count % 10000 == 0){
                    log.info("mysql handle data  number:"+count);
                    // 将数据添加到 bulkProcessor
                    for (HashMap<String,String> hashMap2 : dataList){
                        bulkProcessor.add(new IndexRequest(POSITION_INDEX).source(hashMap2));
                    }
                    // 每提交一次 清空 map 和  dataList
                    map.clear();
                    dataList.clear();
                }
                // 处理 未提交的数据
                for (HashMap<String,String> hashMap2 : dataList){
                    bulkProcessor.add(new IndexRequest(POSITION_INDEX).source(hashMap2));
                }
                bulkProcessor.flush();
            }

        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            try {
                rs.close();
                ps.close();
                connection.close();
                boolean terinaFlag = bulkProcessor.awaitClose(150L,TimeUnit.SECONDS);
                log.info("terinaFlag={}", terinaFlag);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    private BulkProcessor getBulkProcessor(RestHighLevelClient client) {
        BulkProcessor bulkProcessor = null;
        try {
            // 监听
            BulkProcessor.Listener listener = new BulkProcessor.Listener() {
                @Override
                public void beforeBulk(long executionId, BulkRequest bulkRequest) {
                    log.info("beforeBulk-{}", bulkRequest.numberOfActions());
                }

                @Override
                public void afterBulk(long executionId, BulkRequest bulkRequest, BulkResponse bulkResponse) {
                    log.info("afterBulk success-{},id:{}", bulkRequest.numberOfActions(), executionId);
                }

                @Override
                public void afterBulk(long executionId, BulkRequest bulkRequest, Throwable throwable) {
                    log.info("afterBulk fail-{},id:{}", throwable, executionId);
                }
            };

            //
            BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer = (request, bulkListener) -> client
                    .bulkAsync(request, RequestOptions.DEFAULT, bulkListener);

            BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
            builder.setBulkActions(5000);
            builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
            builder.setConcurrentRequests(10);
            builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
            builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
            // 参数设置生效
            bulkProcessor = builder.build();
        } catch (Exception e) {
            e.printStackTrace();
            try {
                bulkProcessor.awaitClose(100L, TimeUnit.SECONDS);
            } catch (Exception e1) {
                log.error(e1.getMessage());
            }
        }
        return bulkProcessor;
    }
}
